package imse.team0.extapp.client.invoice;

import imse.team0.invoice.Invoice;
import imse.team0.invoice.PaymentConfirmation;
import org.apache.activemq.ConnectionFailedException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.jms.*;
import javax.naming.InitialContext;
import javax.xml.bind.JAXBContext;
import javax.xml.bind.JAXBException;
import javax.xml.bind.Marshaller;
import javax.xml.bind.Unmarshaller;
import java.io.*;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author Eugen, 5/4/12 2:12 PM
 */
public class InvoiceReceiver {
    private ExecutorService pool;
    private static final String DESTINATION_QUEUE = "invoicePaymentConfirmation";

    public InvoiceReceiver() {
        pool = Executors.newFixedThreadPool(3);

    }

    public static void main(String[] args) {
        BufferedReader br = new BufferedReader(new InputStreamReader(InvoiceReceiver.class.getResourceAsStream("/client_endpoints.properties")));
        String line;
        try {
            // skip comment lines at the beginning of the file
            while ((line = br.readLine()).startsWith("#")) ;
            final String ip;
            int port;
            int portIncrement;
            int startID;
            int endID;
            String tokens[] = line.split("=");
            ip = tokens[1];
            tokens = br.readLine().split("=");
            port = Integer.parseInt(tokens[1]);
            tokens = br.readLine().split("=");
            portIncrement = Integer.parseInt(tokens[1]);
            tokens = br.readLine().split("=");
            startID = Integer.parseInt(tokens[1]);
            tokens = br.readLine().split("=");
            endID = Integer.parseInt(tokens[1]);
            port -= portIncrement;
            for (int i = 0; i <= endID - startID; i++) {
                final InvoiceReceiver receiver = new InvoiceReceiver();
                port += portIncrement;
                final int finalPort = port;
                Thread th = new Thread(new Runnable() {
                    public void run() {
                        receiver.startInvoiceReceiver(ip, finalPort);
                    }
                });
                th.setDaemon(false);
                th.setName("Invoice Receiver: "+(i+1));
                th.start();
            }
            br.close();
            Thread th = new Thread(new CleanupJmsConnectionTask());
            th.setName("My JMS Cleanup Thread");
            Runtime.getRuntime().addShutdownHook(th);
        } catch (IOException ex) {
            System.err.println("client_endpoints.properties could not be read...");
            throw new RuntimeException(ex);
        }
    }

    public void startInvoiceReceiver(String ip, int portNum) {
        try {
            ServerSocket listenSock = new ServerSocket();
            listenSock.setReuseAddress(true);
            listenSock.bind(new InetSocketAddress(ip, portNum), 5);
            System.out.println("Client Invoice Application : listening for invoices at address :: ip=" + ip + ", port=" + portNum);
            while (true) {
                Socket connection = listenSock.accept();
                pool.execute(new InvoiceProcessor(connection, portNum));
            }
        } catch (IOException ex) {
            throw new RuntimeException(ex);
        }
    }
}

//TODO: set the ip and port of the JMS broker to an external instance...
class InvoiceProcessor implements Runnable {

    static Logger LOG = LoggerFactory.getLogger(InvoiceProcessor.class);

    static Connection conn = null;
    static Queue invoicePaymentQueue = null;

    private Marshaller marshaller = null;

    private Socket connection;
    private int portNum;

    private static void getJMSManagedObjects() {
        try {
            InitialContext ctx = new InitialContext();
            QueueConnectionFactory connFactory = (QueueConnectionFactory) ctx.lookup("ConnectionFactory");
            conn = connFactory.createConnection();
            invoicePaymentQueue = (Queue) ctx.lookup("invoicePaymentQueue");
            conn.start();
            LOG.info("Connection to JMS broker established successfully !");
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    static {
        getJMSManagedObjects();
    }

    public InvoiceProcessor(Socket conn, int portNum) {
        connection = conn;
        this.portNum = portNum;

        try {
            JAXBContext context = JAXBContext.newInstance(imse.team0.invoice.PaymentConfirmation.class);
            marshaller = context.createMarshaller();
        } catch (JAXBException e) {
            e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
            throw new RuntimeException(e);
        }
    }

    public void run() {
        String xmlString = null;
        try {
            JAXBContext jc = JAXBContext.newInstance(imse.team0.invoice.Invoice.class);
            Unmarshaller unmarshaller = jc.createUnmarshaller();
            BufferedReader buffIn = new BufferedReader(new InputStreamReader(connection.getInputStream()));
            xmlString = "";
            String line = null;
            while ((line = buffIn.readLine()) != null && !(line.equals(""))) {
                xmlString += line;
                if (line.contains("</invoice>")) break;
            }
            final byte[] ack = "ACK\r\n".getBytes();
            OutputStream out = connection.getOutputStream();
            out.write(ack);
            out.flush();
            connection.close();
            ByteArrayInputStream input = new ByteArrayInputStream(xmlString.getBytes());
            Invoice invoice = (Invoice) unmarshaller.unmarshal(input);
            // ! Here we must send the payment confirmation to the bank
            // but since this would be just another indirection step, we just send
            // the payment confirmation ourselves directly to the queue
            sendPaymentConfirmation(invoice);
            //System.out.println("Received invoice: " + invoice.getOrderId() +
            //        "\n\twith additional code=" + invoice.getInvoiceGroup() + "\n\twith totalCost=" + invoice.getTotalCost());
            //TODO: receive invoice and put an acknowledgment of payment into the configured queue
        } catch (Throwable ex) {
            LOG.error("Exception thrown while processing an incoming invoice (Client Port= " + portNum + ")");
            LOG.error("Received following XML from the network:\n$$$\n" + xmlString + "\n$$$");
            ex.printStackTrace();
            Runtime.getRuntime().exit(-1);
        }
    }

    private void sendPaymentConfirmation(Invoice invoice) {
        PaymentConfirmation payment = new PaymentConfirmation();
        payment.orderid = invoice.getOrderId();
        StringWriter out = new StringWriter();
        try {
            marshaller.marshal(payment, out);
        } catch (Exception e) {
            LOG.error("\tCaught exception:" + e);
            LOG.error("\tfor payment.orderid=" + payment.orderid);
            throw new RuntimeException(e);
        }
        try {
            QueueSession session = null;
            try {
                session = (QueueSession) conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
            } catch (ConnectionFailedException ex) {
                // probably in the mean time the JMS broker was shut down (e.g. when I restart SMX but Client Invoice System is still running..)
                synchronized (InvoiceProcessor.class) {
                    getJMSManagedObjects();
                }
                session = (QueueSession) conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
            }
            QueueSender sender = session.createSender(invoicePaymentQueue);
            Message jmsMessage = session.createTextMessage(out.toString());
            jmsMessage.setStringProperty("invoiceGroup", invoice.getInvoiceGroup());
            sender.send(jmsMessage);
            session.close(); // must create sessions each time since
            // different threads might be used to call this code (using a Thread Pool)
            LOG.info("Sent message with invoiceGroup header =" + invoice.getInvoiceGroup() + " and orderid=" + payment.orderid + " to the queue");
        } catch (JMSException e) {
            e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
            //throw new RuntimeException(e);
        }
    }
}

class CleanupJmsConnectionTask implements Runnable {

    public void run() {
        InvoiceProcessor.LOG.info("Shutting down the connection to the JMS Broker...");
        if (InvoiceProcessor.conn != null)
            try {
                InvoiceProcessor.conn.close();
            } catch (JMSException e) {
                e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
                throw new RuntimeException(e);
            }
        InvoiceProcessor.LOG.info("All cleanup tasks where completed successfully!");
    }
}
